package com.sxkiler.demo.config;


import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSONObject;
import com.sxkiler.demo.strategy.RERocketMqStrategy;
import com.sxkiler.demo.utils.LogUtil;
import com.sxkiler.demo.utils.SpringContextUtil;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import java.lang.reflect.Method;
import java.util.List;
import java.util.Map;

@Component
@Configuration
public class RocketMqConsumerConfig {
    private static Logger logger = LoggerFactory.getLogger(RocketMqConsumerConfig.class);

    @Value("${rocketmq.namesrvAddr:172.20.178.59:9876}")
    private String namesrvAddr;

    @Value("${rocketmq.vipChannelEnable:false}")
    private String vipChannelEnable;

    @Value("${rocketmq.consumer.groupname:spring-test-group}")
    private String groupName;

    @Value("${rocketmq.consumer.threadMin:20}")
    private String consumeThreadMin;

    @Value("${rocketmq.consumer.threadMax:120}")
    private String consumeThreadMax;

    @Value("${rocketmq.consumer.timeout:1}")
    private String consumeTimeout;

    @Value("${rocketmq.consumer.topics:spring-test-topic}")
    private String topics;


    public static ConcurrentHashSet<String> msgSet= new ConcurrentHashSet<>();

    private static DefaultMQPushConsumer consumer = null;

    public static DefaultMQPushConsumer getConsumer(){
        return consumer;
    }

    /**
     * 创建消息消费的实例
     * @return
     * @throws MQClientException
     */
    @Bean
    public DefaultMQPushConsumer pushConsumer() throws MQClientException {
        if(consumer==null){
            consumer = new DefaultMQPushConsumer(groupName);
        }
        consumer.setNamesrvAddr(namesrvAddr);
        //是否走vipchannel，如果是需要开通10909端口
        consumer.setVipChannelEnabled(vipChannelEnable!=null?Boolean.valueOf(vipChannelEnable):false);
        consumer.setConsumeThreadMin(Integer.parseInt(consumeThreadMin));
        consumer.setConsumeThreadMax(Integer.parseInt(consumeThreadMax));
//        consumer.setPullBatchSize(2);
        consumer.setConsumeTimeout(Long.parseLong(consumeTimeout));
//        consumer.setAllocateMessageQueueStrategy(new RERocketMqStrategy());
        String[] topicList = this.topics.split(",");
        for(String topic:topicList){
            consumer.subscribe(topic,"*");
        }
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                    List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                String msgId = "";
                String body = "";
                String adaptor = "";
                try {
                    if (msgs.size() == 0)
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    for (MessageExt msg : msgs) {
                        msgId = msg.getMsgId();
                        body = new String(msg.getBody(),"utf-8");

                        logger.info("================{},{},{},{},{}",LogUtil.getThreadId(),msg.getQueueId(),msg.getQueueOffset(),msg.getMsgId(), body);
                        //解析参数,获取adaptor
                        Map params = null;
                        //增加判断,兼容原来的发送方式
                        params= (Map)JSONObject.parseObject(body);
                        //获取adaptor
                        adaptor = (String)params.get("adaptor");
                        int index = adaptor.lastIndexOf(".");
                        String bean_name = adaptor.substring(0, index);
                        String method = adaptor.substring(index+1);
                        //将传过来的值放在mdc中
                        //2. 找到适配器处理消息
                        Object biz = SpringContextUtil.getBean(bean_name);
                        Class clazz = biz.getClass();
                        Method m1 = clazz.getDeclaredMethod(method, Map.class);
                        //判断是否进行处理
//                        if(msgSet.contains(msgId)) {
//                            logger.info("================{},{},{},{},重复消费为完成",msg.getQueueId(),msg.getQueueOffset(),msg.getMsgId(), body);
//                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//                        }
//                        msgSet.add(msgId);
                        LogUtil.startTransaction();
                        //反射调用消息处理方法
                        m1.invoke(biz,params);
                        LogUtil.endTransaction();
                        //执行完成,打印结束日志
                        logger.info("================{},{},{},{},消费完成",msg.getQueueId(),msg.getQueueOffset(),msg.getMsgId(), body);
                    }
                } catch (Exception e) {
                    logger.error("RocketMq消费异常,adaptor="+adaptor+",msgId="+msgId+","+e.getMessage(),e);
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }finally {
                    //清空MDC
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        return consumer;
    }



}
